import getopt
import sys
import time
import traceback
from multiprocessing import Pool

import pandas as pd
from elasticsearch.exceptions import ConnectionTimeout, NotFoundError

from air_web.config.config import config
from air_web.data_platform import init_db
from air_web.web_flask.dal.base_dal import EsBaseDal


def sync_day_data(start_time, end_time):
    print(f"start {start_time},{end_time}")
    index_name = config.get("ACLR_WEATHER_HOUR", "aclr_weather_hour")

    t1 = time.time()
    es_dal = EsBaseDal(config["ES_HOST"])
    rules = [
        ("datetime", "query", ">=", start_time),
        ("datetime", "query", "<", end_time),
    ]
    es_res = es_dal.ruleng_query(
        rules,
        index_name=index_name,
        doc_time_field="datetime",
        time_out=300,
    )

    data_list = []
    for r in es_res:
        data_list.append(r)
    if len(data_list) == 0:
        print(f"数据为空,index:{index_name},rule:{start_time},{end_time}")
        return
    t2 = time.time()
    sql_engine.update_data_by_id(data_list, "aclr_weather_hour")
    t3 = time.time()
    print(
        f"数据条数:{len(data_list)}, {start_time},{end_time}, time:{t2 - t1},{t3 - t2}"
    )


def main():
    t1 = time.time()
    date_list = pd.date_range(
        start=start_date, end=end_date, freq="1D"
    ).strftime("%Y-%m-%d")

    pro_pool = Pool(process_num)
    results = []
    for i in range(len(date_list) - 1):
        start_time = date_list[i]
        end_time = date_list[i + 1]
        sync_day_data(start_time, end_time)
        results.append(
            pro_pool.apply_async(
                func=sync_day_data,
                args=(start_time, end_time),
            )
        )
    try:
        [res.get() for res in results]
    except:
        print(traceback.format_exc())

    pro_pool.close()
    pro_pool.join()
    t2 = time.time()
    print(f"end, time:{t2 - t1}")


if __name__ == "__main__":
    sql_engine = init_db(config["MYSQL_DB_NAME"])

    start_date = None
    end_date = None
    process_num = 3

    opts, args = getopt.getopt(sys.argv[1:], "s:e:p:i:d")
    for opt, val in opts:
        if opt == "-s":
            start_date = val
        elif opt == "-e":
            end_date = val
        elif opt == "-p":
            process_num = int(val)

    main()
